package prictice.trident.functions;

import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;

public class Split extends BaseFunction {
    private static final long serialVersionUID = 8207915767656423116L;
    private String patton = null;

    public Split(String patton) {
        this.patton = patton;
    }

    @Override
    public void execute(TridentTuple tuple, TridentCollector collector) {
        String sentence = tuple.getString(0);
        for (String word : sentence.split(patton)) {
            collector.emit(new Values(word));
        }
    }
}